package trouger.dmt;

class SelfAggregationRecordGroupFilter
implements RecordGroupFilter{

	RecordOutputer recordOutputer;
	FieldIndexPair[] resultFieldsIndex;
	AggregationOperation[] aggregationOperation;
	AggregationOperator[] aggregationOperator;
	String[] aggregatedRecord;
	
	public void setResultFieldsIndex(FieldIndexPair[] resultFieldsIndex){
		this.resultFieldsIndex = resultFieldsIndex;
	}
	public void setAggregationOperations(AggregationOperation[] aggregationOperation){
		if (aggregationOperation == null){
			throw new IllegalArgumentException("aggregation operation must be specified for a SelfAggregationRecordGroupFilter");
		}
		this.aggregationOperation = aggregationOperation;
		int n = aggregationOperation.length;
		aggregationOperator = new AggregationOperator[n];
		for (int i = 0; i < n; i++){
			try {
				aggregationOperator[i] = aggregationOperation[i].getOperatorClass().newInstance();
			} catch (Exception e) {
				throw new IllegalArgumentException("on creating AggregationOperator instance, " + e.toString());
			} 
			int srcIdx = aggregationOperation[i].getSourceField().getSourceIndex();
			if (srcIdx > 1){
				throw new IllegalStateException("self aggregation cannot be done on a source table with index greater than one");
			}
		}
	}
	
	@Override
	public boolean beginRecordSource(int sourceIndex) {
		if (sourceIndex != 1){
			throw new IllegalArgumentException("source-index not equals one when doing self aggregation");
		}
		return true;
	}

	@Override
	public boolean endRecordSource() {
		int n = resultFieldsIndex.length;
		Object[] resultRecord = new Object[n + 1];
		for (int i = 0; i < n; i++){
			int j = resultFieldsIndex[i].getSourceIndex();
			if (j != 1){
				throw new IllegalArgumentException("source table index not equal to one in a self-aggregation job");
			}
			int k = resultFieldsIndex[i].getColumnIndex();
			resultRecord[i + 1] = this.aggregatedRecord[k];
		}
		int i = 0;
		for (AggregationOperation op : aggregationOperation){
			resultRecord[op.getResultField()] = aggregationOperator[i++].getAggregatedValue();
		}
		recordOutputer.output(resultRecord);
		return true;
	}
	
	@Override
	public void endOfGroup() {
		// TODO Auto-generated method stub
		
	}

	@Override
	public boolean processRecord(String[] record) {
		this.aggregatedRecord = record;
		int n = aggregationOperation.length;
		int m = record.length;
		for (int i = 0; i < n; i++){
			FieldIndexPair field = aggregationOperation[i].getSourceField();
			int k = field.getColumnIndex();
			this.aggregationAcceptValue(i, k < m ? record[k] : null);
		}
		return true;
	}
	
	protected void aggregationAcceptValue(int aggIndex, String value){
		try{
			aggregationOperator[aggIndex].acceptValue(value);
		}catch(RuntimeException e){
			if (aggregationOperation[aggIndex].isResumeOnError()){
				System.err.println("Aggregation: On error resume next,");
				e.printStackTrace(System.err);
			}else{
				throw e;
			}
		}
	}

	@Override
	public void setRecordOutputer(RecordOutputer recordOutputer) {
		this.recordOutputer = recordOutputer;
	}


}
